package com.zhang.gmall.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhang.gmall.beans.TableProcess;
import com.zhang.gmall.common.GmallConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

/**
 * @title: 业务数据 维度数据动态分流
 * @author: zhang
 * @date: 2022/3/23 17:09
 */
public class TableProcessFunc extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    private Connection connection;

    public TableProcessFunc(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    //处理广播数据
    @Override
    public void processBroadcastElement(String value, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
        //1 解析数据
        JSONObject jsonObject = JSON.parseObject(value);
        String data = jsonObject.getString("data");
        TableProcess tableProcess = JSON.parseObject(data, TableProcess.class);

        //2 校验表是否存在，如果不存在则创建表
        checkTable(
                tableProcess.getSinkTable(),
                tableProcess.getSinkPk(),
                tableProcess.getSinkColumns(),
                tableProcess.getSinkExtend()
        );

        //3 将数据写入状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        broadcastState.put(tableProcess.getSourceTable(), tableProcess);
    }

    private void checkTable(String sinkTable, String sinkPk, String sinkColumns, String sinkExtend) {
        PreparedStatement statement = null;
        try {
            if (sinkPk == null || "".equals(sinkPk)) {
                sinkPk = "id";
            }
            if (sinkExtend == null) {
                sinkExtend = "";
            } else {
                sinkExtend = "," + sinkExtend;
            }

            StringBuilder createTableSQL = new StringBuilder("CREATE TABLE IF NOT EXISTS ")
                    .append(GmallConfig.HBASE_SCHEMA)
                    .append(".").append(sinkTable).append("( ");
            String[] fields = sinkColumns.split(",");
            for (int i = 0; i < fields.length; i++) {
                String field = fields[i];
                //判断是否为主键
                if (field.equals(sinkPk)) {
                    createTableSQL.append(field).append(" VARCHAR PRIMARY KEY ");
                } else {
                    createTableSQL.append("indo").append(".").append(field).append(" VARCHAR ");
                }
                if (i < fields.length - 1) {
                    createTableSQL.append(",");
                }
            }
            //使用 Phoenix 创建表的时候，需要设置 COLUMN_ENCODED_BYTES 属性为 0，即不让 Phoenix 对 column family 进行编码
            //默认column family 为 0
            createTableSQL.append(" ) ").append(" column_encoded_bytes=0 ").append(sinkExtend);
            System.out.println("Phoenix建表语句：" + createTableSQL);
            statement = connection.prepareStatement(createTableSQL.toString());
            statement.execute();
        } catch (SQLException e) {
            e.printStackTrace();
            throw new RuntimeException("创建表：" + sinkTable + "失败！");
        } finally {
            if (statement != null) {
                try {
                    statement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }


    //处理业务数据
    @Override
    public void processElement(JSONObject value, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //4 获取状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        TableProcess tableProcess = broadcastState.get(value.getString("table"));

        String type = value.getString("type");
        if (tableProcess != null && ("bootstrap-start".equals(type) || "insert".equals(type) || "update".equals(type))) {
            //根据配置过滤字段
            filterColumns(tableProcess.getSinkColumns(), value.getJSONObject("data"));
            //3.写出数据
            //为数据打上目的地标签
            value.put("sinkTable", tableProcess.getSinkTable());
            out.collect(value);
        } else {
            System.out.println("Key:" + value.getString("table") + "不存在！");
        }
    }

    @Override
    public void close() throws Exception {
        connection.close();
    }

    private void filterColumns(String sinkColumns, JSONObject data) {
        String[] split = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(split);
        //        Iterator<Map.Entry<String, Object>> iterator = data.entrySet().iterator();
        //        while (iterator.hasNext()) {
        //            Map.Entry<String, Object> next = iterator.next();
        //            if (!columnList.contains(next.getKey())) {
        //                iterator.remove();
        //            }
        //        }
        data.entrySet().removeIf(next -> !columnList.contains(next.getKey()));
    }
}
